package pageRank;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;

import mapred.CombineConfig;
import mapred.InputConfig;
import mapred.Job;
import mapred.MapConfig;
import mapred.OutputConfig;
import mapred.ReduceConfig;

import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.io.compress.GzipCodec;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.RunningJob;
import org.apache.hadoop.mapred.SequenceFileInputFormat;
import org.apache.hadoop.mapred.SequenceFileOutputFormat;

public class Readable extends Job {

	private int round;
	private String input;
	private String output;

	public Readable(String input, String output, int round, String[] settings) {
		super("Readable " + round, settings);
		this.round = round;
		this.input = input;
		this.output = output;
		
	}

	@Override
	protected CombineConfig getCombineConfig() {
		return CombineConfig.newInstance(null);
	}

	@Override
	protected InputConfig getInputConfig(FileSystem fs) throws IOException {
		return InputConfig.newInstance(SequenceFileInputFormat.class,
				input + "/PageRank_" + round);
	}

	@Override
	protected String getLogPath() {
		return "/user/jwLiu/CJKPageRank/logs";
	}

	@Override
	protected MapConfig getMapConfig() {
		return MapConfig.newInstance(Readable_Mapper.class, 0, BytesWritable.class,
				BytesWritable.class, null);
	}

	@Override
	protected OutputConfig getOutputConfig(FileSystem fs) throws IOException {
		return OutputConfig.newInstance(SequenceFileOutputFormat.class, BytesWritable.class,
				BytesWritable.class, output, GzipCodec.class);
	}

	@Override
	protected ReduceConfig getReduceConfig() {
		return ReduceConfig.newInstance(null, 0);
	}

	@Override
	protected void setUp(JobConf conf, FileSystem fs) throws IOException {
		SequenceFileOutputFormat.setOutputCompressionType(conf, SequenceFile.CompressionType.BLOCK);
		conf.set("dfs.http.address", "10.251.2.1:50070");
		double z = Double.parseDouble(new BufferedReader(new InputStreamReader(
				fs.open(new Path("Z_" + round)))).readLine());
		long websize = Long.parseLong(new BufferedReader(new InputStreamReader(
				fs.open(new Path("WebSize_" + round)))).readLine());
		conf.set("Z", String.valueOf(z * 0.85 / websize));
	}

	@Override
	protected void tearDown(JobConf conf, FileSystem fs, RunningJob job)
			throws IOException {
	}

}
